package sanshui.system.trade.factory.trade.strategy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sanshui.system.trade.factory.data.Bar;
import sanshui.system.trade.factory.trade.TradeSystemContext;
import sanshui.system.trade.factory.trade.indicator.IndicatorResult;

import java.util.*;
import java.util.concurrent.*;

/**
 * 多线程处理同一种指标
 */
public class StrategyExecutor {
    private static final Logger logger = LoggerFactory.getLogger(StrategyExecutor.class);

    ExecutorService threadPool;
    ExecutorCompletionService<List<IndicatorResult>> service;

    private int numThread = 10; // 默认单线程

    TradeStrategy codeHisStrategy; // 缓存
    List<Future<List<IndicatorResult>>> futureTaskList = new ArrayList<>();
    List<IndicatorResult> indicatorResults = new ArrayList<>();
    Map<String, TradeStrategy> targetStrategyMap;

    public StrategyExecutor(TradeStrategy strategy) {
        this.codeHisStrategy = strategy;
        threadPool = Executors.newFixedThreadPool(numThread);
        service = new ExecutorCompletionService<>(threadPool);
    }

    public StrategyExecutor(int numThread, Map<String, TradeStrategy> targetStrategyMap) {
        this.numThread = numThread;
        this.targetStrategyMap = targetStrategyMap;
        threadPool = Executors.newFixedThreadPool(numThread);
        service = new ExecutorCompletionService<>(threadPool);
    }

    /**
     * 多线程选股
     */
    public void submit(Map<String, Bar> stockInfoVoMap, TradeSystemContext tradeSystemContext) {
        // 按照线程数量对需要指标计算的标的进行分配
        int fromIndex = 0;
        int size = stockInfoVoMap.keySet().size();
        List<Integer> pageNum = new ArrayList<>();
        int perThreadCodeNum = size / this.numThread;
        int extraAdd = size % this.numThread;

        for (int i = 0; i < this.numThread; i++) {
            if (extraAdd == 0) {
                pageNum.add(perThreadCodeNum);
            } else {
                pageNum.add(perThreadCodeNum + 1);
                --extraAdd;
            }
        }

        if (this.targetStrategyMap == null || this.targetStrategyMap.keySet().size() == 0) {
            logger.info("没有标的需要进行多线程选股");
            return;
        }

        List<String> targetsTemp = new ArrayList<>(targetStrategyMap.keySet());
        futureTaskList = new ArrayList<>();
        // 把标的分到各个线程中
        for (int i = 0; i < this.numThread; i++) {
            List<String> pageCodes = targetsTemp.subList(fromIndex, fromIndex + pageNum.get(i));
            if (!pageCodes.isEmpty()) {
                Map<String, TradeStrategy> indicatorMapPage = targetIndicatorMapPage(targetStrategyMap, pageCodes);
                Callable<List<IndicatorResult>> futureTask = executeIndicator(indicatorMapPage, stockInfoVoMap, tradeSystemContext);
                Future<List<IndicatorResult>> submit = service.submit(futureTask);
                futureTaskList.add(submit);
            }
            fromIndex += pageNum.get(i);
        }
    }

    private Callable<List<IndicatorResult>> executeIndicator(Map<String, TradeStrategy> strategyMapPage, Map<String, Bar> stockInfoVoMap, TradeSystemContext tradeSystemContext) {
        return () -> {
            List<IndicatorResult> results = new ArrayList<>();
            for (String code : strategyMapPage.keySet()) {
                results.add(strategyMapPage.get(code).exec(stockInfoVoMap.get(code), tradeSystemContext));
            }
            return results;
        };
    }

    public List<IndicatorResult> takeResults() {
        if (futureTaskList.isEmpty()) {
            return new ArrayList<>();
        }
        List<IndicatorResult> result = new ArrayList<>();
        try {

            Iterator<Future<List<IndicatorResult>>> iterator = futureTaskList.iterator();
            while (iterator.hasNext()) {
                Future<List<IndicatorResult>> next = iterator.next();
                try {
                    result = next.get(5000L, TimeUnit.SECONDS);
                } catch (Exception e) {
                    logger.error("获取并发选股出错", e);
                    throw new IllegalStateException("并发选股出错, 原因：线程出错");
                }

                if (result == null) {
                    logger.error("结果未空");
                    throw new IllegalStateException("并发选股出错, 原因：结果未空");
                }

                indicatorResults.addAll(result);
                iterator.remove();
            }
            return indicatorResults;
        } catch (Exception e) {
            logger.error("获取并发结果失败", e);
        }
        indicatorResults.addAll(result);
        return indicatorResults;
    }

    public void close() {
        if (!threadPool.isShutdown()) {
            threadPool.isShutdown();
        }
        service = null;
        codeHisStrategy = null;
        futureTaskList = null;
        indicatorResults = null;
    }


    private Map<String, TradeStrategy> targetIndicatorMapPage(Map<String, TradeStrategy> targetIndicatorMap, List<String> pageCodes) {
        Map<String, TradeStrategy> targetIndicatorMapPage = new HashMap<>();
        for (String code : pageCodes) {
            targetIndicatorMapPage.put(code, targetIndicatorMap.get(code));
        }
        return targetIndicatorMapPage;
    }
}
